package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 指数秒级窗口函数
 */
public class IndexSecWindowFunction extends RichWindowFunction<CleanBean, IndexBean, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<IndexBean> out) throws Exception {
        /**
         * 开发步骤：
         * 1.记录最新指数
         * 2.格式化日期
         * 3.封装输出数据
         */
        CleanBean line = null;
        for (CleanBean cleanBean : input) {
            if(line == null){
                line = cleanBean;
            }
            if(line.getEventTime()< cleanBean.getEventTime()){
                line = cleanBean;
            }
        }


        /**
         * 加盐/随机数+ 唯一标识 + 时间戳 (此方法不满足连续查询的业务需求)
         * 唯一标识(证券代码) + 时间戳 （本项目用此方法）
         * startkey = 300059+ 20210825093000  =30005920210825093000 (起始时间)
         * endkey = 300059+ 20210825095000  = 300059202108250935000 (结束时间)
         *
         * 假设查询601375,从早上9:30到11:00的连续秒级行情数据
         * 开始时间: 60137520230729093000
         * 结束时间: 60137520230729110000
         * Scan scan = new Scan()
         * scan.setStartRow()
         * scan.setEndRow()
         */
        String tradeDate = DateUtil.longTimeToString(line.getEventTime(), Constant.format_yyyy_mm_dd);

        Long tradeTime = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        //3.封装输出数据
        IndexBean indexBeanNew = new IndexBean(
                tradeTime,
                line.getSecCode(),
                line.getSecName(),
                line.getPreClosePrice(),
                line.getOpenPrice(),
                line.getHighPrice(),
                line.getLowPrice(),
                line.getTradePrice(),
                0l, 0l,
                line.getTradeVolumn(),
                line.getTradeAmt(),
                tradeDate,
                line.getEventTime(),
                line.getSource());
        out.collect(indexBeanNew);

    }
}
